package hbasecli.controller;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;

import hbasecli.config.DefaultConnectionConfig;
import hbasecli.connection.ConnectionInfo;
import hbasecli.connection.HBaseConnections;
import hbasecli.model.json.param.CreateConnectionParam;
import hbasecli.model.json.JsonResponse;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.bind.annotation.*;

import static hbasecli.model.json.JsonResponse.success;
import static hbasecli.model.json.JsonResponse.failure;
import static org.springframework.util.StringUtils.hasText;

@RestController
public class ConnectionController {

    private static final Logger log = LoggerFactory.getLogger(ConnectionController.class);

    @Resource
    private DefaultConnectionConfig config;

    @Resource
    private HBaseConnections connections;

    private static final Set<String> paramNameSet = new HashSet<>(Arrays.asList(
            "hbase.zookeeper.property.clientPort",
            "hbase.master.port",
            "hbase.zookeeper.quorum",
            "hbase.regionserver.port"
    ));

    @GetMapping("/api/defaultConnectionParams")
    JsonResponse<List<CreateConnectionParam>> getDefaultConnectionParams() {
        List<CreateConnectionParam> data = new ArrayList<>();
        if (hasText(config.getZookeeperPropertyClientPort())) {
            CreateConnectionParam param = new CreateConnectionParam();
            param.setName("hbase.zookeeper.property.clientPort");
            param.setValue(config.getZookeeperPropertyClientPort());
            data.add(param);
        }
        if (hasText(config.getMasterPort())) {
            CreateConnectionParam param = new CreateConnectionParam();
            param.setName("hbase.master.port");
            param.setValue(config.getMasterPort());
            data.add(param);
        }
        if (hasText(config.getZookeeperQuorum())) {
            CreateConnectionParam param = new CreateConnectionParam();
            param.setName("hbase.zookeeper.quorum");
            param.setValue(config.getZookeeperQuorum());
            data.add(param);
        }
        if (hasText(config.getRegionServerPort())) {
            CreateConnectionParam param = new CreateConnectionParam();
            param.setName("hbase.regionserver.port");
            param.setValue(config.getRegionServerPort());
            data.add(param);
        }
        return success(data);
    }

    @PostMapping("/api/connections")
    JsonResponse<ConnectionInfo> createConnection(@RequestBody List<CreateConnectionParam> params) {
        Configuration configuration = new Configuration();
        Map<String, String> usefulParams = new HashMap<>();
        for (CreateConnectionParam param : params) {
            String name = param.getName();
            if (paramNameSet.contains(name)) {
                String value = param.getValue();
                configuration.set(name, value);
                usefulParams.put(name, value);
            }
        }
        Connection connection;
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            log.error("创建连接时出现异常: ", e);
            return failure("创建连接时出现异常:" + e.getMessage());
        }
        if (connection == null) {
            return failure("创建连接失败");
        }
        String name = configuration.get("hbase.zookeeper.quorum");
        if (!hasText(name)) {
            name = "connection:" + System.currentTimeMillis();
        }
        return success(connections.put(connection, usefulParams, name));
    }

    @GetMapping("/api/connections")
    JsonResponse<List<ConnectionInfo>> getAllConnection() {
        return success(connections.list());
    }

    @GetMapping("/api/connections/{id}")
    JsonResponse<ConnectionInfo> getConnectionById(@PathVariable("id") String id) {
        return success(connections.getConnectionInfo(id));
    }

    @DeleteMapping("/api/connections/{id}")
    JsonResponse<ConnectionInfo> deleteConnectionById(@PathVariable("id") String id) {
        return success(connections.remove(id));
    }

}
